/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.trainer;

import java.net.URI;
import maxent.mr.io.ObjectWriter;
import maxent.mr.io.ReadDataTaggedUtil;
import maxent.mr.mapred.FeatureMapper;
import maxent.mr.mapred.FeatureReducer;
import maxent.mr.writable.FeatureKeyWritable;
import maxent.mr.writable.FeatureKeysWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import stanford.tagger.ReadDataTagged;
import stanford.tagger.TrainerExperiments;

/**
 *
 * @author hadoop
 */
public class FeatureRunner extends Configured implements Tool{

    ReadDataTagged rdt;
    TrainerExperiments prob;

    public FeatureRunner(ReadDataTagged rdt){
        this.rdt = rdt;
    }

    public FeatureRunner(TrainerExperiments prob){
        this.prob = prob;
        this.rdt = prob.getReadDataTagged();
    }

    
    public int run(String[] args) throws Exception {

        String inputDir = args[0];
        String outputDir = args[1];
        String distCachePath = "/user/hadoop/temp";

        // write for input dir
        ReadDataTaggedUtil rdtUtil = new ReadDataTaggedUtil(rdt);
        rdtUtil.write(inputDir);


        Configuration conf = new Configuration();
        
        // setting distributed cache for ReadDataTagged
        Path temp = ObjectWriter.write(conf, prob, distCachePath, "prob.dist");
        // Register the file in the DC.  Open the local file "targets"
        DistributedCache.addCacheFile(new URI(temp+"#targets"), conf);
        DistributedCache.createSymlink(conf);
           

        // setting bigger heap space
        //conf.set( "mapred.child.java.opts", "-Xmx2000m");

        Job job = new Job(conf);
        job.setJarByClass(DictionaryRunner.class);

        SequenceFileInputFormat.addInputPath(job, new Path(inputDir));
        SequenceFileOutputFormat.setOutputPath(job, new Path(outputDir));

        job.setMapperClass(FeatureMapper.class);
        job.setReducerClass(FeatureReducer.class);

        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(FeatureKeyWritable.class);

        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(FeatureKeysWritable.class);

        return job.waitForCompletion(true) ? 0 : 1;


    }

    

}
